import multiprocessing
import asyncio
import queue
from typing import Any, List
from loguru import logger
from multiprocessing import Queue
from .services.live_stream import DouyinLiveStream
from .services.consumer import ConsumerPool
from .config.live_config import LiveConfig


class ProcessManager:
    def __init__(self) -> None:
        """初始化进程管理器"""
        self.config = LiveConfig()
        self.processes: List[multiprocessing.Process] = []

    @staticmethod
    async def _run_task_in_process(q: Queue,
                                shared_list: list,
                                shared_var: Any,
                                shared_proxy: Any) -> None:
        """在进程中运行的异步任务"""
        try:
            # 创建线程共享队列
            shared_queue = queue.Queue()
            
            # 初始化直播流处理器和消费者
            live_stream = DouyinLiveStream(shared_proxy)
            consumer = ConsumerPool(shared_queue=shared_queue)
            
            while True:  # 添加无限循环
                try:
                    # 创建任务
                    tasks = [
                        asyncio.create_task(live_stream.run(q, shared_queue, shared_list)),
                        asyncio.create_task(asyncio.to_thread(consumer.run,shared_var))
                    ]
                    
                    # 等待任务完成
                    await asyncio.gather(*tasks)
                    
                except Exception as e:
                    logger.error(f"任务执行出错，准备重试: {str(e)}")
                    await asyncio.sleep(5)
            
        except Exception as e:
            logger.error(f"进程任务运行错误: {str(e)}")

    @staticmethod
    def _process_target(q: Queue,
                       shared_list: list,
                       shared_var: Any,
                       shared_proxy: Any) -> None:
        """
        进程目标函数,用于启动异步事件循环
        """
        try:
            # 获取或创建事件循环
            try:
                loop = asyncio.get_event_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
            
            
            # 运行异步任务
            loop.run_until_complete(
                ProcessManager._run_task_in_process(q, shared_list, shared_var, shared_proxy)
            )
        except Exception as e:
            logger.error(f"进程启动错误: {str(e)}")
        finally:
            loop.close()

    def start_processes(self,
                       q: Queue,
                       shared_list: list,
                       shared_var: Any,
                       shared_proxy: Any) -> None:
        """启动多个进程"""
        logger.info(f"开始启动 {self.config.PROCESS_COUNT} 个处理进程...")
        
        self.processes = [
            multiprocessing.Process(
                target=self._process_target,
                args=(q, shared_list, shared_var, shared_proxy)
            )
            for _ in range(self.config.PROCESS_COUNT)
        ]
        
        for i, process in enumerate(self.processes, 1):
            process.start()
            logger.info(f"进程 {i}/{self.config.PROCESS_COUNT} 已启动，PID: {process.pid}")

    async def stop_processes(self) -> None:
        """异步停止所有进程"""
        for process in self.processes:
            try:
                process.terminate()
                await asyncio.get_event_loop().run_in_executor(None, process.join)
                logger.info(f"Process {process.pid} stopped")
            except Exception as e:
                logger.error(f"停止进程 {process.pid} 失败: {str(e)}")
        self.processes.clear() 